package com.kingsoft.dc.khaos.module.spark.util

import java.text.SimpleDateFormat
import java.util.Date

import com.kingsoft.dc.khaos.module.spark.constants.MetaDataConstants
import com.kingsoft.dc.khaos.util.Logging
import org.apache.commons.lang3.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.sql.types._
import java.util.regex.Pattern

import com.kingsoft.dc.khaos.KhaosContext

import scala.collection.mutable.ArrayBuffer
import scala.util.control.Breaks._

/**
  *
  * create by gs 19-10-23
  *
  */
object DataTypeConvertUtils extends Logging {
  var kc: KhaosContext = null
  //日期时间类型枚举convertDataLine
  private val defaultDate = new SimpleDateFormat("yyyy-MM-dd") //默认
  private val date1 = new SimpleDateFormat("yyyy-MM-dd") //默认
  private val date2 = new SimpleDateFormat("yyyy/MM/dd")
  private val date3 = new SimpleDateFormat("yyyyMMdd")
  private val date4 = new SimpleDateFormat("yyyy年MM月dd日")
  //日期类型枚举
  private val defaultDatetime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") //默认
  private val datetime1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") //默认
  private val datetime2 = new SimpleDateFormat("yyyyMMddHHmmss")
  private val datetime3 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS")
  private val datetime4 = new SimpleDateFormat("yyyyMMddHHmmssSSS")
  private val datetime5 = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
  private val datetime6 = new SimpleDateFormat("MM-dd HH:mm:ss")
  private val datetime7 = new SimpleDateFormat("MMddHHmmss")
  private val datetime8 = new SimpleDateFormat("yyyy年MM月dd日 HH时mm分ss秒")
  //  private val datetime9 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss EE")
  //  private val datetime9 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss zzz")
  //  private val datetime9 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss yyy")

  def convertDatetype2StringType(df: DataFrame): DataFrame = {
    var df1 = df
    for (column <- df.columns) {
      df1 = df1.withColumn(column, df1.col(column).cast(StringType))
    }
    df1
  }

  /**
    * 将RDD行映射为带schema的Row
    *
    * @param schema
    * @param rdd
    * @param delimiter
    * @return
    */
  def rdd2Rows(schema: StructType, rdd: RDD[Array[String]], delimiter: String = " "): RDD[Row] = {
    rdd.map(attributes => {
      var row: Row = Row()
      try {
        row = convertDataLine(schema, attributes)
      } catch {
        case e => {
          throw new Exception(s"=>>> 数据转换失败! line=${attributes.mkString(delimiter)}", e)
        }
      }
      row
    })
  }

  /**
    * 带忽略错误转换
    *
    * @param schema
    * @param attributes
    * @param delimiter
    * @return
    */
  def checkDataLineAbnormal(schema: StructType, attributes: Array[String], delimiter: String = " ") = {
    try {
      convertDataLine(schema, attributes)
      true
    } catch {
      case e: Exception => {
        e.printStackTrace()
        false
      }
    }
  }

  def addStringType(sourceRow: Row, value: String) = {
    var row = sourceRow
    if (MetaDataConstants.NULL == value || StringUtils.isBlank(value)) {
      row = Row.merge(row, Row(null))
    } else {
      row = Row.merge(row, Row(value))
    }
    row
  }

  def addLongType(sourceRow: Row, value: String) = {
    var row = sourceRow
    if (MetaDataConstants.NULL == value || StringUtils.isBlank(value)) { //数据中包含NULL值
      row = Row.merge(row, Row(null))
    } else {
      row = Row.merge(row, Row(value.trim.toLong))
    }
    row
  }

  def addDateType(sourceRow: Row, value: String) = {
    var row = sourceRow
    if (MetaDataConstants.NULL == value || StringUtils.isBlank(value)) {
      row = Row.merge(row, Row(null))
    } else {
      row = Row.merge(row, Row(getDate(value.trim)))
    }
    row
  }

  def addTimestampType(sourceRow: Row, value: String) = {
    var row = sourceRow
    if (MetaDataConstants.NULL == value || StringUtils.isBlank(value)) {
      row = Row.merge(row, Row(null))
    } else {
      row = Row.merge(row, Row(getDateTime(value.trim)))
    }
    row
  }

  def addDoubleType(sourceRow: Row, value: String) = {
    var row = sourceRow
    if (MetaDataConstants.NULL == value || StringUtils.isBlank(value)) {
      row = Row.merge(row, Row(null))
    } else {
      row = Row.merge(row, Row(value.trim.toDouble))
    }
    row
  }

  def addDecimalType(sourceRow: Row, value: String): Row = {
    var row = sourceRow
    if (MetaDataConstants.NULL == value || StringUtils.isBlank(value)) {
      row = Row.merge(row, Row(null))
    } else {
      row = Row.merge(row, Row(new java.math.BigDecimal(value.trim)))
    }
    row
  }

  def addIntType(sourceRow: Row, value: String): Row = {
    var row = sourceRow
    if (MetaDataConstants.NULL == value || StringUtils.isBlank(value)) {
      row = Row.merge(row, Row(null))
    } else {
      row = Row.merge(row, Row(value.trim.toInt))
    }
    row
  }

  /**
    * 转换数据行为Row
    *
    * @param schema
    * @param attributes
    * @return
    */
  def convertDataLine(schema: StructType, attributes: Array[String]) = {
    var row: Row = Row()
    for (i <- 0 until schema.size) {
      val colType = schema(i).dataType
      row = colType match {
        case StringType => addStringType(row, attributes(i))
        case LongType => addLongType(row, attributes(i))
        case DateType => addDateType(row, attributes(i))
        case TimestampType => addTimestampType(row, attributes(i))
        case DoubleType => addDoubleType(row, attributes(i))
        case DecimalType() => addDecimalType(row, attributes(i))
        case NullType => Row.merge(row, Row(attributes(i)))
        case _ => Row.merge(row, Row(attributes(i)))
      }
    }
    row
  }


  /*  /**
     * 转换数据行为Row
     *
     * @param schema
     * @param attributes
     * @return
     */
    def convertDataLine(schema: StructType, attributes: Array[String]) = {
      var row: Row = Row()
      for (i <- 0 until schema.size) {
        val colType = schema(i).dataType
        colType match {
          case StringType => {
            if (MetaDataConstants.NULL == attributes(i) || StringUtils.isBlank(attributes(i))) {
              row = Row.merge(row, Row(null))
            } else {
              row = Row.merge(row, Row(attributes(i)))
            }
          }
          case LongType => {
            if (MetaDataConstants.NULL == attributes(i) || StringUtils.isBlank(attributes(i))) { //数据中包含NULL值
              row = Row.merge(row, Row(null))
            } else {
              row = Row.merge(row, Row(attributes(i).trim.toLong))
            }
          }
          case DateType => {
            if (MetaDataConstants.NULL == attributes(i) || StringUtils.isBlank(attributes(i))) {
              row = Row.merge(row, Row(null))
            } else {
              row = Row.merge(row, Row(getDate(attributes(i).trim)))
            }
          }
          case TimestampType => {
            if (MetaDataConstants.NULL == attributes(i) || StringUtils.isBlank(attributes(i))) {
              row = Row.merge(row, Row(null))
            } else {
              row = Row.merge(row, Row(getDateTime(attributes(i).trim)))
            }
          }
          case DoubleType => {
            if (MetaDataConstants.NULL == attributes(i) || StringUtils.isBlank(attributes(i))) {
              row = Row.merge(row, Row(null))
            } else {
              row = Row.merge(row, Row(attributes(i).trim.toDouble))
            }
          }
          case NullType => row = Row.merge(row, Row(attributes(i)))
          case _ => row = Row.merge(row, Row(attributes(i)))
        }
      }
      row
    }*/

  def defaultDateFormat(date: Date): Date = {
    java.sql.Date.valueOf(defaultDate.format(date))
  }

  def defaultDatetimeFormat(date: Date): Date = {
    java.sql.Timestamp.valueOf(defaultDatetime.format(date))
  }

  /**
    * 统一时间格式
    *
    * @param dateStr
    * @return
    */
  /*  def getDate(dateStr: String): Date = {
      if (isValidDateTime(dateStr, date1)) {
        defaultDateFormat(date1.parse(dateStr))

      } else if (isValidDateTime(dateStr, date2)) {
        defaultDateFormat(date2.parse(dateStr))

      } else if (isValidDateTime(dateStr, date3)) {
        defaultDateFormat(date3.parse(dateStr))

      } else if (isValidDateTime(dateStr, date4)) {
        defaultDateFormat(date4.parse(dateStr))
      } else {
        throw new Exception(s"=>不支持该date类型转换！dateStr=[${dateStr}]")
      }
    }*/

  /**
    * 统一日期时间格式
    *
    * @param datetimeStr
    * @return
    */
  /*def getDateTime(datetimeStr: String): Date = {
    if (isValidDateTime(datetimeStr, datetime1)) {
      defaultDatetimeFormat(datetime1.parse(datetimeStr))

    } else if (isValidDateTime(datetimeStr, datetime2)) {
      defaultDatetimeFormat(datetime2.parse(datetimeStr))

    } else if (isValidDateTime(datetimeStr, datetime3)) {
      defaultDatetimeFormat(datetime3.parse(datetimeStr))

    } else if (isValidDateTime(datetimeStr, datetime4)) {
      defaultDatetimeFormat(datetime4.parse(datetimeStr))

    } else if (isValidDateTime(datetimeStr, datetime5)) {
      defaultDatetimeFormat(datetime5.parse(datetimeStr))

    } else if (isValidDateTime(datetimeStr, datetime6)) {
      defaultDatetimeFormat(datetime6.parse(datetimeStr))

    } else if (isValidDateTime(datetimeStr, datetime7)) {
      defaultDatetimeFormat(datetime7.parse(datetimeStr))

    } else if (isValidDateTime(datetimeStr, datetime8)) {
      defaultDatetimeFormat(datetime8.parse(datetimeStr))

    } else {
      throw new Exception(s"=>不支持该datetime类型转换！datetimeStr=[${datetimeStr}]")
    }
  }*/

  /**
    * 校验日期时间格式
    *
    * @param dateStr
    * @return
    */
  def isValidDateTime(dateStr: String, format: SimpleDateFormat): Boolean = {
    var convertSuccess = true
    try {
      format.setLenient(false)
      format.parse(dateStr)
    } catch {
      case e => {
        logError(s"=> convert datetime type error!")
        logError(s"=> 失败原因：${e.getCause}")
        logError(s"=> 失败信息：${e.getMessage}")
        convertSuccess = false
      }
    }
    convertSuccess
  }

  /**
    * 统一时间格式
    *
    * @param dateStr
    * @return
    */
  def getDate(dateStr: String): Date = {
    val date = getDateAndDateTime(dateStr)
    defaultDateFormat(date)
    //    java.sql.Date.valueOf(defaultDate.format(date))
  }

  /**
    * 统一日期时间格式
    *
    * @param datetimeStr
    * @return
    */
  def getDateTime(datetimeStr: String): Date = {
    val datetime = getDateAndDateTime(datetimeStr)
    defaultDatetimeFormat(datetime)
    //    java.sql.Timestamp.valueOf(defaultDatetime.format(datetime))
  }

  /**
    *
    *
    * @param dateStr
    * @return
    */
  def getDateAndDateTime(dateStr: String): Date = {
    val dateformat = new SimpleDateFormat(getDateFormat(dateStr))
    dateformat.parse(dateStr)
  }

  /**
    * 根据日期数据内容获取日期格式
    *
    * @param str
    * @return
    */
  def getDateFormat(str: String): String = {
    var year: Boolean = false
    val pattern: Pattern = Pattern.compile("^[-\\+]?[\\d]*$")
    if (StringUtils.isNotBlank(str) && pattern.matcher(str.substring(0, 4)).matches()) {
      year = true
    }
    val ab = new ArrayBuffer[Char](str.length)
    var index: Int = 0
    if (!year) {
      if (str.contains("月") || str.contains("-") || str.contains("/")) {
        if (Character.isDigit(str.charAt(0))) {
          index = 1
        }
      } else {
        index = 3
      }
    }
    for (i <- 0 until str.length) {
      val chr = str.charAt(i);
      if (Character.isDigit(chr)) {
        if (index == 0) {
          ab.append('y')
        }
        if (index == 1) {
          ab.append('M')
        }
        if (index == 2) {
          ab.append('d')
        }
        if (index == 3) {
          ab.append('H')
        }
        if (index == 4) {
          ab.append('m')
        }
        if (index == 5) {
          ab.append('s')
        }
        if (index == 6) {
          ab.append('S')
        }
      } else {
        if (i > 0) {
          val lastChar = str.charAt(i - 1)
          if (Character.isDigit(lastChar)) {
            index = index + 1
          }
        }
        ab.append(chr)
      }
    }
    val format = ab.mkString match {
      case "yyyyyyyy" => "yyyyMMdd"
      case "yyyyyyyyyy" => "yyyyMMddHH"
      case "yyyyyyyyyyyy" => "yyyyMMddHHmm"
      case "yyyyyyyyyyyyyy" => "yyyyMMddHHmmss"
      case _ => ab.mkString
    }
    format
  }
}
